--set hive.mapred.mode ='proceed'
with tab_bagging as (
    select t1.waybill_no,
           t1.scan_site_code,
           t1.scan_site,
           t1.scan_time,                         --建包扫描时间
           t1.scan_user_code,                    --建包扫描人员编码
           t1.scan_user,                         --建包扫描人员名称
           t1.package_code as package_code,      --建包包号
           t2.package_code as bagging_first_code --建包实际一段码
    from (
             select waybill_no,
                    scan_site_code,
                    scan_site,
                    scan_time,
                    package_code,
                    scan_user_code,
                    scan_user,
                    network_type,
                    row_number() over(partition by waybill_no, scan_site_code order by scan_time desc) rk
             from jms_dwd.dwd_tab_barscan_bagging_base_hi
             where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -3) and date_add('{{ execution_date | cst_ds }}',1)
               and (network_type= 3 or network_type= 4)
               and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
         ) t1
             left join
         (
             select package_code, package_number
             from jms_dwd.dwd_electronic_package_list_base_hi
             where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -15) and date_add('{{ execution_date | cst_ds }}',1)
               and is_used !=0                       --'包牌使用状态 0未使用 1已使用 2已拆包'
                    and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
             group by package_code, package_number
         ) t2 on t1.package_code = t2.package_number
    where t1.rk = 1
      and t1.scan_time between concat(date_add('{{ execution_date | cst_ds }}', -2), ' 00:00:00') and concat(date_add('{{ execution_date | cst_ds }}',1), ' 02:59:59')

),
     tab_waybill as (
         select waybill_no                           , --运单号
                ordersource_code order_source_code,                                   ---订单来源编码
                ordersource_name order_source_name,                                   --订单来源名称
                goods_type_code,                                     --商品类型
                goods_type_name,
                cast(package_length as decimal(16,2))    package_length , --包裹长
                cast(package_wide as decimal(16,2))      package_wide,  --包裹宽
                cast(package_high as decimal(16,2))      package_high,  --包裹高
                package_inside_charge_weight as package_charge_weight,  --包裹计费重
                customer_code,                                       --客户名称
                customer_name,                                       --客户编码
                dispatch_network_code,                               ---派件网点
                dispatch_network_name,                               --派件网点名称
                package_inside_charge_weight      --内部计费重
         from jms_dwd.dwd_yl_oms_oms_waybill_incre_hi
         where substr(dt, 1, 10) BETWEEN date_add ('{{ execution_date | cst_ds }}', -30) and date_add('{{ execution_date | cst_ds }}',1)
     ),

     tab_order as  --订单表取三段码和实际揽收网点
         (
             select
                 t.waybill_id
                  ,t.terminal_dispatch_code
                  ,t.first_code
                  ,f.center_code as end_center_code
                  ,f.center_name as end_center_name
                  ,t.real_pick_network_code
                  ,net1.name as real_pick_network_name
             from
                 (
                     select   --订单表取实际实际取件网点和三段码
                         orders.waybill_id
                          ,orders.terminal_dispatch_code
                          ,split(orders.terminal_dispatch_code,',')[0] as first_code
                          ,nvl(real_pick_network_code,pick_network_code) as real_pick_network_code
                     from jms_dwd.dwd_yl_oms_oms_order_dt orders
                     where substr(dt, 1, 10) BETWEEN date_add ('{{ execution_date | cst_ds }}', -30) and date_add('{{ execution_date | cst_ds }}',0)
                       and date(input_time) between date_add ('{{ execution_date | cst_ds }}', -30) and date_add('{{ execution_date | cst_ds }}',0)
union all
select   --订单表取实际实际取件网点和三段码
    orders.waybill_id
     ,orders.terminal_dispatch_code
     ,split(orders.terminal_dispatch_code,',')[0] as first_code
     ,nvl(real_pick_network_code,pick_network_code) as real_pick_network_code
from jms_dwd.dwd_yl_oms_oms_order_hf orders
where dt between concat(date_add('{{ execution_date | cst_ds }}',1),'-00') and concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
  and substr(input_time,1,13) between concat(date_add('{{ execution_date | cst_ds }}',1) ,' 00') and concat(date_add('{{ execution_date | cst_ds }}',1),' 02' )
  and waybill_rnk=1
    )t left join (select code,center_code,center_name from  jms_dim.dim_lmdm_sys_first_code where is_enable=1 and is_delete =1 group by code,center_code,center_name) f on f.code = t.first_code
    left join jms_dim.dim_network_whole_massage net1 on t.real_pick_network_code=net1.code

    ),
    tab_difficult as
    (
select waybill_no       --运单
from jms_dwd.dwd_tab_barscan_difficult_base_hi
where substr(dt, 1, 10) >=date_add('{{ execution_date | cst_ds }}', -7)
  and problem_type_code in ('23b','23c')
  and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
group by waybill_no
    ),

--差异化建包表
    package_different as (
select
    express_org_code, --网点编码
    express_center_short_code, --一段码
    update_time
from
    (
    select
    express_org_code, --网点编码
    express_center_short_code, --一段码
    update_time,
    row_number() over(partition by express_org_code order by update_time desc ) rk
    from jms_dim.dim_yl_sys_package_different_base
    )t where rk=1
    ),
--差异化建包表 一段码是否有变更
    package_different2 as (
select
    express_center_short_code code, --一段码
    update_time
from  jms_dim.dim_yl_sys_package_different_base --and express_org_code='0571025'
    ),
--一段码表是否有变更
    first_update as (
select
    code, --一段码
    update_time                --变更时间
from  jms_dim.dim_lmdm_sys_first_code
    ),

--网点省市区对应的一段码表
    network_first as (
select
    a.code, --网点编码
    collect_set(b.code) as  fisrt_codes
from jms_dim.dim_network_whole_massage a
    inner join (select * from jms_dim.dim_lmdm_sys_first_code where is_enable=1 and destination_province_id is not null and destination_city_id is not null and destination_area_id is not null) b
on ( a.provider_id=b.destination_province_id and a.city_id=b.destination_city_id and a.area_id=b.destination_area_id)
group by a.code
    ),
--实际扫描的出仓网点
    tab_delivery as (
select /*+BROADCAST(p),BROADCAST(net)*/
     t.waybill_no,
    t.scan_site_code,
    t.scan_site,
    t.scan_time,
    p.express_center_short_code,  --网点在差异化建包表中对应的一段码
    net.fisrt_codes               --派件网点省市区对应的一一段码
from (
    select waybill_no,
    scan_site_code,
    scan_site,
    scan_time,
    row_number() over(partition by waybill_no order by scan_time) rk
    from jms_dwd.dwd_tab_barscan_deliver_base_hi
    where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -5) and date_add('{{ execution_date | cst_ds }}',4)
    and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
    ) t
    left join package_different p on p.express_org_code=t.scan_site_code
    left join network_first net on net.code=t.scan_site_code
where rk = 1
    ),
--退转件面单
    tab_reback as (
select *from (
    select t1.waybill_no,                                                    --运单号
    t1.print_flag,                                                    --面单是否已打印
    t1.terminal_dispatch_code,                                        --原三段码  以 ,为分割
    t1.new_terminal_dispatch_code,                                   --新三段码  以 ' '为分割
    ''                                           as dispatch_code_rek,--退转件中的派件网点
    t1.abnormal_reg_network_id,--问题件登记网点
    t1.abnormal_reg_network_name,--问题件登记网点名称
    split(t1.terminal_dispatch_code, ',')[0]     as first_code_old,   --一段码
    split(t1.new_terminal_dispatch_code, ' ')[0] as first_code_new,    --一段码
    print_time,
    row_number() over(partition by t1.waybill_no order by print_time) as rk
    from jms_dwd.dwd_tab_reback_transfer_express_base_hi t1
    where
    substr(dt,1,10) >= date_add('{{ execution_date | cst_ds }}', -7)
            --t1.apply_time >= date_add('{{ execution_date | cst_ds }}', -7)
            --and t1.status = '2'  --已审核
    and t1.print_flag='1' --面单已经打印
    )t where rk=1
    ),
    tab_bagging2 as (  --取最晚建包时间
select t1.waybill_no,
    t1.scan_site_code,
    t1.scan_site,
    t1.scan_time,                         --建包扫描时间
    t1.scan_user_code,                    --建包扫描人员编码
    t1.scan_user,                         --建包扫描人员名称
    t1.package_code as package_code,      --建包包号
    t2.package_code as bagging_first_code --建包实际一段码
from (
    select waybill_no,
    scan_site_code,
    scan_site,
    scan_time,
    package_code,
    scan_user_code,
    scan_user,
    row_number() over(partition by waybill_no, scan_site_code order by scan_time desc) rk
    from jms_dwd.dwd_tab_barscan_bagging_base_hi
    where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -5) and date_add('{{ execution_date | cst_ds }}',1)
    and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
    ) t1
    left join
    (
    select package_code, package_number
    from jms_dwd.dwd_electronic_package_list_base_hi
    where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -15) and date_add('{{ execution_date | cst_ds }}',1)
    and is_used !=0                       --'包牌使用状态 0未使用 1已使用 2已拆包'
    and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
    group by package_code, package_number
    ) t2 on t1.package_code = t2.package_number
where t1.rk = 1
    ),
--实际扫描的签收网点（出库+签收）
    tab_sign as (
select tt.waybill_no,
    tt.scan_site_code,
    tt.scan_site,
    tt.scan_time
from (
    select waybill_no,
    scan_site_code,
    scan_site,
    scan_time,
    post_station_code,
    real_oper_source_code,
    row_number() over(partition by waybill_no order by scan_time) as rk
    from (
    select waybill_no,                           --运单
    signed_site_code as scan_site_code,   --出库网点
    signed_site      as scan_site,        --出库网点name
    signed_time      as scan_time,
    null             as post_station_code, --驿站编码
    null             as real_oper_source_code --实操作来源编码
    from jms_dwd.dwd_tab_barscan_sign_base_hi
    where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -2) and date_add('{{ execution_date | cst_ds }}',7)
    and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
    union ALL
    select waybill_no,       --运单
    scan_site_code,   --出库网点
    scan_site,        --出库网点name
    scan_time,
    post_station_code, --驿站编码
    real_oper_source_code
    from jms_dwd.dwd_tab_barscan_collect_base_hi
    where substr(dt, 1, 10) between date_add('{{ execution_date | cst_ds }}', -2) and date_add('{{ execution_date | cst_ds }}',7)
    and scan_type = '出库扫描'
    and dt<=concat(date_add('{{ execution_date | cst_ds }}',1),'-02')
    ) t
    ) tt
where tt.rk = 1
    ),
    tab_update48 as (
select
    waybill_no
from
    (
    select
    waybill_no
    from  tab_bagging a
    inner join  first_update b on a.bagging_first_code=b.code and  cast((unix_timestamp(scan_time)-unix_timestamp(update_time))/60/60/48 as decimal(16,2))<=1
    and cast((unix_timestamp(scan_time)-unix_timestamp(update_time))/60/60/48 as decimal(16,2))>0
    group by waybill_no
    union all
    select
    waybill_no
    from  tab_bagging a
    inner join  package_different2 b on a.bagging_first_code=b.code and  cast((unix_timestamp(scan_time)-unix_timestamp(update_time))/60/60/48 as decimal(16,2))<=1
    and cast((unix_timestamp(scan_time)-unix_timestamp(update_time))/60/60/48 as decimal(16,2))>0
    group by waybill_no
    )t group by waybill_no)
insert overwrite table jms_tmp.tmp_network_bagging_detail_bag_dt partition(dt)
select
    tab_bagging.waybill_no,
    tab_bagging.scan_site_code,
    tab_bagging.scan_site,
    tab_bagging.scan_time as bagging_scan_time,                         --建包扫描时间
    tab_bagging.scan_user_code as bagging_user_code,                    --建包扫描人员编码
    tab_bagging.scan_user as bagging_user,                         --建包扫描人员名称
    tab_bagging.package_code,                      --建包包号
    tab_bagging.bagging_first_code,                --建包实际一段码
    tab_waybill.order_source_code,                                   --订单来源编码
    tab_waybill.order_source_name,                                   --订单来源名称
    tab_order.first_code,                                          --运单三段码
    tab_waybill.goods_type_code,                                     --商品类型
    tab_waybill.goods_type_name,
    tab_waybill.package_length,                                      --包裹长
    tab_waybill.package_wide,                                        --包裹宽
    tab_waybill.package_high,                                        --包裹高
    tab_waybill.package_charge_weight,                               --包裹计费重
    tab_waybill.customer_code,                                       --客户名称
    tab_waybill.customer_name,                                       --客户编码
    tab_waybill.dispatch_network_code,                               --派件网点
    tab_waybill.dispatch_network_name,                               --派件网点名称
    tab_order.real_pick_network_code as pick_network_code,                                   --派件网点
    tab_order.real_pick_network_name as pick_network_name,                                   --派件网点名称
    tab_order.end_center_code,                                       --末端网点
    tab_order.end_center_name,                                       --末端网点
    tab_delivery.scan_site_code as delivery_scan_code,               --扫描出仓网点
    tab_delivery.scan_site as delivery_scan,                         --扫描出仓网点
    tab_delivery.express_center_short_code,                          --派件网点在差异化建包表中对应的一段码
    tab_delivery.fisrt_codes,                                        --派件网点省市区在一段码表中对应的一段码
    tab_sign.scan_site_code as sign_scan_code  ,                     --扫描签收网点
    tab_sign.scan_site as sign_scan  ,
    tab_sign.scan_time as sign_scan_time  ,
    tab_reback.waybill_no as reback_waybill_no,                                           --退转件中运单号
    tab_reback.print_flag,                                           --面单是否已打印
    tab_reback.terminal_dispatch_code,                               --原三段码  以 ,为分割
    tab_reback.new_terminal_dispatch_code,                           --新三段码  以 ' '为分割
    tab_reback.dispatch_code_rek,                                    --退转件中的派件网点
    tab_reback.first_code_old,                                       --一段码
    tab_reback.first_code_new,                                       --一段码
    tab_reback.print_time,
    tab_bagging2.scan_time as bagging_scan_time2,                   --建包扫描最晚时间
    tab_bagging2.bagging_first_code as bagging_first_code2,                   --建包扫描最晚时间
    tab_difficult.waybill_no as  difficult_waybill_code,             --错分类型
    sort_array(array(nvl(tab_waybill.package_length,0),nvl(tab_waybill.package_wide,0),nvl(tab_waybill.package_high,0))) as bian_sort,
    tab_update48.waybill_no as waybill_no48,                          --48小时有变更的一断码对应的单号
    from_unixtime(unix_timestamp(tab_bagging.scan_time)-3*60*60,'yyyy-MM-dd HH:mm:ss') as bagging_scan_time3,
--date(tab_bagging.scan_time)  as dt
    date(from_unixtime(unix_timestamp(tab_bagging.scan_time)-3*60*60,'yyyy-MM-dd HH:mm:ss')) as dt
from tab_bagging
    left join tab_waybill on tab_waybill.waybill_no=tab_bagging.waybill_no
    left join tab_delivery on tab_delivery.waybill_no=tab_bagging.waybill_no
    left join tab_reback  on  tab_reback.waybill_no=tab_bagging.waybill_no
    left join tab_bagging2 on tab_bagging.waybill_no=tab_bagging2.waybill_no
    and tab_bagging.scan_site_code=tab_bagging2.scan_site_code
    left join tab_difficult on tab_difficult.waybill_no=tab_bagging.waybill_no
    left join tab_update48 on tab_update48.waybill_no=tab_bagging.waybill_no
    left join tab_sign on tab_sign.waybill_no=tab_bagging.waybill_no
    left join tab_order on tab_order.waybill_id=tab_bagging.waybill_no
distribute by pmod(hash(waybill_no), 64);
-----------最终明细---------------------
insert overwrite table jms_dm.dm_network_bagging_detail_bag_dt partition(dt)
select
    waybill_no,
    bagging_site_code,
    bagging_site,
    bagging_scan_time,
    bagging_user_code,
    bagging_user,
    package_code,
    bagging_first_code,
    order_source_code,
    order_source_name,
    first_code,
    goods_type_code,
    goods_type_name,
    package_length,
    package_wide,
    package_high,
    package_charge_weight,
    customer_code,
    customer_name,
    dispatch_network_code,
    dispatch_network_name,
    pick_network_code,
    pick_network_name,
    end_center_code,
    end_center_name,
    delivery_scan_code,
    delivery_scan,
    express_center_short_code,
    fisrt_codes,
    sign_scan_code ,
    sign_scan,
    sign_scan_time,
    reback_waybill_no,
    print_flag,
    terminal_dispatch_code,
    new_terminal_dispatch_code,
    dispatch_code_rek,
    first_code_old,
    first_code_new,
    print_time,
    bagging_scan_time2,
    bagging_first_code2,
    difficult_waybill_code,
    bian_sort,
    waybill_no48,
    bagging_scan_time3,
    is_small,
    is_slender,
    is_can_err,
    case --若快件尚未签收&出仓派件，则判定为正确建包
        when first_code is null and sign_scan_code is null and delivery_scan_code is null then 0
        --若快件已签收&出仓派件，且签收&出仓网点在差异化建包表中，查询该网点在差异化建包表的一段码与建包扫描一段码是否一致，不一致为错建包
        when is_can_err=1 and first_code is null and express_center_short_code is not null and bagging_first_code is not null and  express_center_short_code!=bagging_first_code then 1
        --若快件已签收&出仓派件，但签收&出仓网点不在差异化建包表中，根据该网点的省市区匹配一段码表，获取一段码，判断与建包扫描一段码是否一致，不一致为错建包
        when is_can_err=1 and first_code is null and express_center_short_code is null and fisrt_codes is not null and bagging_first_code is not null and array_contains(fisrt_codes,bagging_first_code)
            then 1
        --运单不为退转件或者退转件面单未打印，则判断运单一段码与建包扫描一段码是否一致，一致则为正确建包，不一致为错建包
        when is_can_err=1 and first_code is not null and  reback_waybill_no is null and bagging_first_code is not null and first_code!=bagging_first_code then 1
        --运单为退转件并且面单已打印，建包扫描时间若早于面单打印时间，判断原单一段码与建包扫描一段码是否一致，一致则为正确建包
        when is_can_err=1 and first_code is not null and  reback_waybill_no is not null and bagging_scan_time<print_time and bagging_first_code is not null and first_code_old is not null and first_code_old!=bagging_first_code then 1
        --运单为退转件并且面单已打印，建包扫描时间若晚于面单打印时间，判断新单一段码与建包扫描一段码是否一致，一致则为正确建包，不一致为错建包
        when is_can_err=1 and first_code is not null and  reback_waybill_no is not null and bagging_scan_time2>print_time and bagging_first_code2 is not null and first_code_new is not null and  first_code_new!=bagging_first_code2 then 1
        else 0
        end as is_err,  --是否错建包（1错建0否）
    is_special,
    is_lightweight,
    substr(bagging_scan_time3,1,10) as dt
from
    (
        select
            *,
            if(
--生鲜 液体 易碎品 文件、普通物品类型
                    (nvl(PACKAGE_LENGTH,0) <= 50 and nvl(PACKAGE_WIDE,0) <= 50 and nvl(PACKAGE_HIGH,0) <= 50 --边长不超过30cm
                        and  PACKAGE_CHARGE_WEIGHT <= 3 --所有物品类型<=3
                        and nvl(PACKAGE_LENGTH,0) + nvl(PACKAGE_WIDE,0)+ nvl(PACKAGE_HIGH,0) <= 90
                        )
                ,1,0) as is_small, --是否是小件
            if(nvl(PACKAGE_CHARGE_WEIGHT,0) <=3 and bian_sort[2]<=70 and cast(bian_sort[2]/bian_sort[1] as decimal(16,2))>=5 and PACKAGE_LENGTH+PACKAGE_WIDE+PACKAGE_HIGH<=90
                ,1,0) as is_slender,--是否是细长件
            if(
                            nvl(PACKAGE_CHARGE_WEIGHT,0) <=3
                        and nvl(PACKAGE_LENGTH,0) <= 80 and nvl(PACKAGE_WIDE,0) <= 80 and nvl(PACKAGE_HIGH,0) <= 80
                        and nvl(PACKAGE_LENGTH,0) + nvl(PACKAGE_WIDE,0)+ nvl(PACKAGE_HIGH,0) <= 100
                        and nvl(PACKAGE_LENGTH,0)* nvl(PACKAGE_WIDE,0)* nvl(PACKAGE_HIGH,0)*1000000<=0.03
                ,1,0) as is_special, --是否是特殊件
            if(
                            nvl(PACKAGE_CHARGE_WEIGHT,0) <=0.03
                        and nvl(PACKAGE_LENGTH,0) <= 10 and nvl(PACKAGE_WIDE,0) <= 10 and nvl(PACKAGE_HIGH,0) <= 1
                ,1,0) as is_lightweight --是否轻薄件
                ,if(waybill_no48 is  null and  difficult_waybill_code is null ,1,0) as is_can_err
        from jms_tmp.tmp_network_bagging_detail_bag_dt
--bidefault.dm_network_bagging_detail_bag_dt_tmp
        where dt between date_add('{{ execution_date | cst_ds }}',-2) and '{{ execution_date | cst_ds }}'
          and substr(waybill_no,1,1)!='B'
    ) t1
    distribute by pmod(hash(waybill_no), 64);